package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class LogSplitProcessFunction extends ProcessFunction<String,String> {

    OutputTag<String> errTag;
    OutputTag<String> startTag;
    OutputTag<String> displayTag;
    OutputTag<String> actionTag;

    // 构造方法传入实参
    public LogSplitProcessFunction(OutputTag<String> errTag, OutputTag<String> startTag, OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errTag = errTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String s, Context context, Collector<String> collector) throws Exception {
        // 1-解析JSON数据
        JSONObject jsonObject = JSON.parseObject(s);
        // 2-错误日志
        if (jsonObject.get("err") != null) {
            // 侧流错误日志
            context.output(errTag,jsonObject.toJSONString());
            jsonObject.remove("err");
        }
        // 2-启动日志
        if (jsonObject.get("start") != null) {
            // 侧流启动日志
            context.output(startTag,jsonObject.toJSONString());
            // 3-页面日志
        } else {
            // 获取曝光和动作数据
            JSONArray displays = jsonObject.getJSONArray("displays");
            JSONArray actions = jsonObject.getJSONArray("actions");
            // 删除曝光和动作数据
            jsonObject.remove("displays");
            jsonObject.remove("actions");
            // 主流页面日志
            collector.collect(jsonObject.toJSONString());
            if (displays != null && !displays.isEmpty()) {
                for (Object display : displays) {
                    // 添加曝光数据
                    jsonObject.put("display",display);
                    // 侧流曝光日志
                    context.output(displayTag,jsonObject.toJSONString());
                }
                // 删除曝光数据
                jsonObject.remove("display");
            }
            if (actions != null && !actions.isEmpty()) {
                for (Object action : actions) {
                    // 添加动作日志
                    jsonObject.put("action",action);
                    // 侧流动作日志
                    context.output(actionTag,jsonObject.toJSONString());
                }
            }
        }
    }
}
